package com.example.demo.controller;

import com.example.demo.componet.MapUtil;
import com.example.demo.componet.ReadCSV;
import com.example.demo.componet.WordCountMapBean;
import com.example.demo.componet.spark.ReadCSVReturnRDD;
import com.example.demo.model.BrandJson;
import com.example.demo.model.GoodsJson;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import scala.Tuple2;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/analyze")
public class DisplayGoodsController {


    @Resource
    private JavaSparkContext sc;

    //文件地址
    @Autowired
    private String getFilePath;



    //sse实时推送对象服务
    ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

    private List<String> getGoodsList() {
        ReadCSV readCSV = new ReadCSV();

        //List<String> goodsList =
         return  readCSV.getColumnList(getFilePath, 5);   //"C:\\Users\\fireapproval\\Desktop\\毕业设计\\数据集\\test.csv"
    }
    @GetMapping("/goods")
    public GoodsJson display1(){

        List<String> tmpList = getGoodsList().subList(1,flag);

        flag = flag + 1;

        //内存中加载数据

        JavaRDD<String> rdd = sc.parallelize(tmpList);

        //JavaRDD<String> rdd1 = rdd.map(x -> x.concat("xxx"));

        //转化为(word,1)类型
        JavaPairRDD<String, Integer> wordMap = rdd.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));

        //对(word,1)计算  //必须要用匿名函数

        JavaPairRDD<String, Integer> reduceMap = wordMap.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);


        //map接收rdd
        Map<String, Integer> rddMap = reduceMap.collectAsMap();

        //排序
        Map<String, Integer> sortMap = MapUtil.sortDescend(rddMap);

                 /*
                解析数据，遍历map
                 */

        List<String> nameList = new ArrayList<>();
        List<Integer> valueList = new ArrayList<>();
        GoodsJson goodsJson = new GoodsJson();
        //计数器
        int index=0;
        for (Map.Entry<String,Integer> entry:sortMap.entrySet()
        ) {
            if(entry.getKey() !=""){
                nameList.add(entry.getKey());
                valueList.add(entry.getValue());
            }

            index = index+1;
            if (index >=5 ) break;
        }

        goodsJson.setName(nameList);
        goodsJson.setValue(valueList);

        System.out.println("-----goods data-----:"+goodsJson.toString());

        return  goodsJson;
    }


//
//    ReadCSV readCSV = new ReadCSV();
//
//    List<String> goodsList = readCSV.getColumnList("C:\\Users\\fireapproval\\Desktop\\毕业设计\\数据集\\test.csv", 5);

    //计数器
    int flag =6;

    @GetMapping("/goods1")
    public SseEmitter method1(){

        SseEmitter sseEmitter = new SseEmitter();

        executorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {

                List<String> tmpList = getGoodsList().subList(1,flag);

                flag = flag + 1;

                //内存中加载数据

                JavaRDD<String> rdd = sc.parallelize(tmpList);

                //JavaRDD<String> rdd1 = rdd.map(x -> x.concat("xxx"));

                //转化为(word,1)类型
                JavaPairRDD<String, Integer> wordMap = rdd.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));

                //对(word,1)计算  //必须要用匿名函数

                JavaPairRDD<String, Integer> reduceMap = wordMap.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);


                //map接收rdd
                Map<String, Integer> rddMap = reduceMap.collectAsMap();

                //排序
                Map<String, Integer> sortMap = MapUtil.sortDescend(rddMap);

                 /*
                解析数据，遍历map
                 */

                List<String> nameList = new ArrayList<>();
                List<Integer> valueList = new ArrayList<>();
                GoodsJson goodsJson = new GoodsJson();
                //计数器
                int index=0;
                for (Map.Entry<String,Integer> entry:sortMap.entrySet()
                ) {
                    if(entry.getKey() !=""){
                        nameList.add(entry.getKey());
                        valueList.add(entry.getValue());
                    }

                    index = index+1;
                    if (index >=5 ) break;
                }

                goodsJson.setName(nameList);
                goodsJson.setValue(valueList);

                //System.out.println("-----goods data-----:"+goodsJson.toString());

                try {
                    sseEmitter.send(goodsJson);
                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        },10,5,TimeUnit.SECONDS);


        return sseEmitter;
    }

}
